was not having this ready check a bug before?

Yes, I think so.

My understanding of the API is that this is supposed to have "fire and forget" semantics, where the executors are requested but there's no guarantee that they are up after they have been requested. The return value of the future is only if the request is acknowledged, not if the executors resolved in starting up.

I think you're right. I'm trying to understand if we should then launch each pod and wait for it to go running before continuing in this method. Would that make sense? I think the method is called in a way that there aren't several parallel invocations.

I don't think so - we're returning a future here so even though we don't execute this method multiple times in parallel we might be executing multiple instances of the future at once. Thus we should probably return immediately after creating the pods without waiting for them to become ready. We can have a separate monitor / watch that handles when the executor pods fail to launch.

colocatedPods

Done

executor -> shuffle

Done

the sbin/ folder usually has just binaries in it, maybe put in conf/

Done

Is this the path on the node, on the shuffle pod, or on the executor pod?

This is the path on the executor pod. For now, it's up to the user to ensure that it's consistent with that on the node and the shuffle pod.

I think the rest of Spark uses its local dir instead of the tmpdir for shuffle sans shuffle service?

this logic is very similar to farther below -- maybe extract out a (Pod => Boolean) predicate function called isPodReady that can be used in both places?

Use Readiness.isReady() from the Kubernetes client library.

Use a case class for clarity instead of returning a tuple.

Done

Use Option.map instead of match here. Using map and getOrElse we can use a more fluent builder pattern. Perhaps something like this:

val basePodBuilder = new PodBuilder().<default configuration>.endSpec()
val resolvedPodBuilder = shuffleDir.map { dir =>
  basePodBuilder.editSpec()
    .setVolumes(...)
    .editContainerAt(0).<edits>.endContainer() // Can also use editMatchingContainer()
    .endSpec()
}.getOrElse(basePodBuilder)
kubernetesClient.pods().create(resolvedPod.build())

We could also build the container separately from the pod, but I'm not sure how much that buys us. That would look something like this:

val baseContainerBuilder = new ContainerBuilder().<default configurations>
val basePodBuilder = new PodBuilder().<default configuration>.endSpec()
val resolvedPodBuilder = shuffleDir.map { dir =>
   basePodBuilder.editSpec()
     .setVolumes(...)
     .endSpec()
 }.getOrElse(basePodBuilder)
 val resolvedContainerBuilder = shuffleDir.map { dir =>
   baseContainerBuilder.addToVolumeMounts(...)
 }.getOrElse(baseContainerBuilder)
 val resolvedPod = resolvedPodBuilder().editSpec()
   .addContainer(resolvedContainerBuilder.build())
   .endSpec()
   .build()

Should this command be running the shuffle service? Use bin/spark-class to help here if that's the case.

Fixed

Format seems off

Is the reason of modifying the RetrieveSparkAppConfig is just to get the executorId? You should already get the executorId with RegisterExecutor message.
My 2cents it will be better to have as little modifications possible in Spark core.

The reason is that the driver needs to send slightly different config to each executor - each executor gets a different shuffle service pod IP. The driver needs to differentiate between the each executor when it's fetching config. We can do this at the time of the RegisterExecutor message but would need to extend and override parts of CoarseGrainedExecutorBackend.scala. I suspect the changes will be more complex if we do that.

Is this a official image that we'll be supporting?

+1 to @tnachen's question. Maybe add a comment pointing to the Dockerfile below?

Is this a new configuration you are introducing? We will need documentation for this for sure to know what is this for.

Also please use the ConfigBuilder from config.scala and its .internal() marker if this is not meant to be set by users and is only used internally.

@ash211 do you think we should add it into spark-core alongside spark.shuffle.service.port that already exists there?

I used the configbuilder for references to spark.shuffle.service.host within the Kubernetes package; and left it as it is here, conforming to the surrounding code in spark-core.

Fix the formatting, I don't think this typical Spark style (arg on each line and 4 spaces indented)

Done

Debug only?

looks like setting it lets you override the IP that the executor thinks its colocated shuffle service is on -- should put that in the doc

You should move this back to match Spark style

ditto Style

/tmp might not be ideal in many setups. e.g. It may be mounting small-size memfs not large enough for large shuffle data. It might be better to use a unique dir here like spark-shuffle-dir and document that people should prepare the dir if necessary.

Eventually we should perhaps make this configurable by turning this into a helm chart?

I don't think it would be a helm chart since it doesn't deploy an entire application; just a dependent component (and the whole overhead of helm/tiller would be prohibitive). This is only a sample YAML that a user might use. I was thinking of distributing a YAML file that the user may modify according to their needs.

Trying to understand your points.

it doesn't deploy an entire application

Helm can be used for deploying a service. I consider the spark shuffle service a service independent of individual spark jobs.

Or, by "an entire application", do you mean the fact that the docker image may not be what the user wants?

the whole overhead of helm/tiller would be prohibitive

By "overhead", do you mean the "setup steps" such as downloading helm and doing $ helm init? Or do you mean the "runtime" cpu, etc overhead?

If you mean "setup steps", it may not be a lot compared with the alternative of modifying this yaml file on her own and doing $ kubectl create. At least that was my experience. Another thing to consider is mucking with yaml can easily lead to syntax errors, and troubleshooting hours. Helm chart is a black box shielding inexperienced users from that.

Anyway, I wouldn't block this PR for not having helm. I just think it is nice way to go for the future iterations of this.

Independent of helm or not, I still think putting /tmp here is a not a good example.

One more related point is that we may have multiple dirs in this as an example. Shuffle is seek intensive. Using multiple disks is quite common. We may want to show how it can be done as an example.

By "overhead", do you mean the "setup steps" such as downloading helm and doing $ helm init? Or do you mean the "runtime" cpu, etc overhead?

Yes. Helm is one package manager, but not necessarily installed on most clusters. It would be excessive to expect installations of helm to run a sub-component of Spark in my opinion.

Independent of helm or not, I still think putting /tmp here is a not a good example.

Fair enough, I'll include a better example, of multiple directories then. What do you think would be a good default shuffle directory on hosts? /shuffle-files or something?

Makes sense. You're trying to lower the adoption barrier.

I would think /spark-shuffle-files-0, /spark-shffule-files-1, etc.

Updating with the result of the SIG meeting discussion. We agreed /var/tmp would be good default base dir for this. So maybe /var/tmp/spark-shuffle-files-0, /var/tmp/spark-shuffle-files-1, etc.

+1 to @tnachen's question. Maybe add a comment pointing to the Dockerfile below?

Undo the formatting change here - but IDEs don't handle this well unfortunately.

Move the dot at the end of line 312 to this line

Use the Client class directly as using spark-submit manipulates system properties

Also please use the ConfigBuilder from config.scala and its .internal() marker if this is not meant to be set by users and is only used internally.

I expect this to be set by the user, so remove .internal()

remove .internal()

remove .internal()

when launching say 100 executors at the start of a job and each executor is registering with the driver, this will trigger 100 near-simultaneous requests to the apiserver for running shuffle service pods.

Does it make sense instead to keep a list locally that's updated by a Watch, and to refer to that cached local list instead of querying apiserver each time?

Not sure the impact of this level of query load on apiserver.

I agree with this idea.

That's a good idea. I'll add this caching in

looks like setting it lets you override the IP that the executor thinks its colocated shuffle service is on -- should put that in the doc

something like colocatedPods.map(_.getStatus.getPodIP).mkString(",")

feels like not quite the right match expression here -- how about just _ ?

@mccheah how to match against a 1-item list here? Not sure if it's a Seq or something else at this point

why the nice? should we do this in the other Dockerfiles too?

Thought 0 is the default?

With #227 merged, each component no longer needs to have its own assembly XML file.

Nit: src/main/docker/shuffle -> src/main/docker/shuffle-service for a tad more clarity.

IIUC here doRequestTotalExecutors should return true unless it fails to send the scaling request to the cluster manager. The return value has nothing to do with whether the executors are ready or not.

Both yarn and mesos's backend implementation do something like that.

IMO we should not call k8s api directly inside doRequestTotalExecutors, because the returned Future is waited by the base class's requestExecutors method with a timeout. The current code could block the dynamic allocation manager thread for a long time, e.g. when there is a intermittent network problem, or cause it to exit if there is any exception thrown when calling k8s api. We should use a worker thread to do it, wrap any possible k8s exception, and return a real Future<Boolean> here.

K8s API calls are asynchronous in and of themselves, as long as we're not specifically watching the resources for their readiness. I think it's actually semantically correct to try to request the executors here from the API server. If there's a network timeout or a failure then the state of the Future should reflect as such. YarnSchedulerBackend does this as well, in that it queries the application master with a message indicating the target number of executors that is desired. In client mode this could be a remote call.

There is one difference here I think. Kubernetes will not refuse to accept new pods even if there are no resources, and they'll just go pending. We need some mechanism in between to ensure we do not request more executors than can be accommodated at the current time.

This is also the case in YARN. The request to the application master only sets the desired number of executors: (see this method). The application master will then continuously attempt to fetch these resources after the initial request in its allocator thread.

Thus even if the cluster does not actually have the sufficient resources available, true is still returned by doRequestTotalExecutors in YARN mode.

Ah, I didn't realized the body of doRequestTotalExecutors is actually running in a thread pool provided by the implicit val requestExecutorContext executor.

But the other question still holds: we should return true unless it fails to to contact k8s api.